#!/usr/bin/env python
#-*- coding: UTF-8 -*-
# dispatcher.py: dispatcher implementaion
#
# Copyright (C) 2008-2010 Hily Jiang <hilyjiang@gmail.com>
#
# This file is part of dtask.
#
# This software is provided 'as-is', without any express or implied
# warranty.  In no event will the authors be held liable for any damages
# arising from the use of this software.
#
# Permission is granted to anyone to use this software for any purpose,
# including commercial applications, and to alter it and redistribute it
# freely, subject to the following restrictions:
#
# 1. The origin of this software must not be misrepresented; you must not
#    claim that you wrote the original software. If you use this software
#    in a product, an acknowledgment in the product documentation would be
#    appreciated but is not required.
# 2. Altered source versions must be plainly marked as such, and must not be
#    misrepresented as being the original software.
# 3. This notice may not be removed or altered from any source distribution.

"""
Dispatcher implementaion.
"""

import sys
import threading
import sqlite3
from time import time


class Dispatcher(object):
    """Dispatcher class, it hold a taskid-schid-nodid list."""

    def __init__(self, dbpath=':memory:', clearInterval=600, dispatchTimeout=300):
        """
        dbpath:
            sqlite3 database save path, default ':memory:' means run in memory.
        clearInterval:
            interval of the timer which clear the timeout dispatch
        dispatchTimeout:
            timeout to remove the dispatch record
        """
        self.dbpath = dbpath
        self.clearInterval = clearInterval
        self.dispatchTimeout = dispatchTimeout

        # timers
        self.clearTimer = self.__startClearTimer()

        # database operation implementation resource
        self.dbThreadLock = threading.Semaphore(0)
        self.dbThreadLock.release()
        self.dbOpEvent = threading.Event()
        self.dbReturnEvent = threading.Event()
        self.dbOpFunc = None
        self.dbOpArgs = None
        self.dbOpKwrgs = None
        self.dbReturn = None

        # Because sqlite3 does not support cross-thread connection sharing,
        # so we create an individual thread to handle sqlite3 operations.
        self.dbthread = threading.Thread(target=self.__dbThreadFunc)
        self.dbthread.setDaemon(True)
        self.dbthread.start()

    def __dbThreadFunc(self):
        """Thread function for sqlite3 operations."""
        self.conn = sqlite3.connect(self.dbpath)
        self.cur = self.conn.cursor()
        try:
            # create table for dispatcher
            self.cur.execute('''
                CREATE TABLE dispatches(
                    taskid INTEGER,
                    schid INTEGER,
                    nodid INTEGER,
                    dtime INTEGER
                )
            ''')
            self.conn.commit()
        except Exception, e: # table exists
            self.cur.execute('DELETE FROM dispatches')
            self.conn.commit()
            print '[DPC]', e

        while True:
            self.dbOpEvent.wait()
            self.dbOpEvent.clear()
            self.dbReturn = self.dbOpFunc(self, *self.dbOpArgs,
                                          **self.dbOpKwargs)
            self.dbReturnEvent.set()

    def __dbThread(func):
        """Decoration function to wrap functions that operate sqlite3.
           The functions decorated will then executed in the db thread."""
        def wrapFunc(self, *args, **kwargs):
            self.dbThreadLock.acquire()
            self.dbOpFunc = func
            self.dbOpArgs = args
            self.dbOpKwargs = kwargs
            self.dbOpEvent.set()
            self.dbReturnEvent.wait()
            self.dbReturnEvent.clear()
            self.dbThreadLock.release()
            return self.dbReturn
        return wrapFunc

    def __startClearTimer(self):
        """Start a timer to check timeout of dispatches ."""
        timer = threading.Timer(self.clearInterval, self.__clearTimerFunc)
        timer.setDaemon(True)
        timer.start()

    def __clearTimerFunc(self):
        """Clear dispatch function. It is intervally executed."""
        self.__clearTimeoutDispatches()
        self.__startClearTimer()

    @__dbThread
    def __clearTimeoutDispatches(self):
        """Check if dispatch has timeout, remove it."""
        nowTime = int(time())
        sql = 'DELETE FROM dispatches WHERE dtime<?'
        self.cur.execute(sql, (nowTime-self.dispatchTimeout,))
        self.conn.commit()

    @__dbThread
    def dispatch(self, schId, taskId, nodeId):
        """Dispatch a new task to a new node from a scheduler"""
        taskId = int(taskId)
        schId = int(schId)
        nodeId = int(nodeId)
        sql = 'INSERT INTO dispatches VALUES(?, ?, ?, ?)'
        self.cur.execute(sql, (taskId, schId, nodeId, int(time())))
        self.conn.commit()

    @__dbThread
    def getSchedulerId(self, schId, taskId, nodeId):
        """Get the scheduler ID which dispach the specified task
           to the specified node."""
        schId = int(schId) # taskId make be the same from different sch
        taskId = int(taskId)
        nodeId = int(nodeId)
        self.cur.execute('SELECT schid FROM dispatches ' \
                         ' WHERE taskid=? AND nodid=?', (taskId, nodeId))
        rs = self.cur.fetchone()
        if rs == None:
            return None
        else:
            return rs[0]

    @__dbThread
    def removeDispatch(self, schId, taskId):
        """Remove a dispatch from list."""
        self.cur.execute('DELETE FROM dispatches' \
                         ' WHERE schid=? AND taskid=?', (schId, taskId))
        self.conn.commit()


def test():
    from time import sleep
    dpc = Dispatcher(clearInterval=6, dispatchTimeout=3)
    while True: sleep(1)

if __name__ == '__main__':
    test()
